package main

import (
	"bufio"
	"io"
	"net"

	"../common"
	m20 "github.com/metrics20/go-metrics20/carbon20"
	// "sync"
	// "time"
	//"fmt"
	"strings"
)

type Receiver struct {
	listener       *net.TCPListener
	udpConn        *net.UDPConn
	Counter        *Counter
	Cache          *common.CacheManager
	readBufferSize int
	validateLevel  m20.ValidationLevelM20
	validateOrder  bool
	DataPointCh    chan common.DataPoint
	metricMapping  map[string]string
}

func NewReceiver(listener *net.TCPListener, udpConn *net.UDPConn, conf *Config,
	counter *Counter, cache *common.CacheManager) *Receiver {
	ch := make(chan common.DataPoint, conf.DataBufferSize)
	r := Receiver{
		listener:       listener,
		udpConn:        udpConn,
		Counter:        counter,
		Cache:          cache,
		readBufferSize: conf.ReadBufferSize,
		validateLevel:  conf.ValidateLevel,
		validateOrder:  conf.ValidateOrder,
		DataPointCh:    ch,
		metricMapping:  make(map[string]string),
	}
	for _, s := range conf.MetricNameConvertions {
		ss := strings.Split(s, "=>")
		if len(ss) == 2 {
			r.metricMapping[ss[0]] = ss[1]
		}
	}
	if r.readBufferSize < 4096 {
		r.readBufferSize = 4096
	}
	return &r
}

func (r *Receiver) Stop() {
	r.listener.Close()
	//stop by sender, ensure all data sent
	close(r.DataPointCh)
}

func (r *Receiver) Run() {
	if r.udpConn != nil {
		go r.handle(r.udpConn)
	}
	for {
		c, err := r.listener.AcceptTCP()
		if nil != err {
			log.Errorf("setup tcp connection, error %v", err)
			break
		}
		go r.handle(c)
	}
}

func (r *Receiver) handle(c net.Conn) {
	defer c.Close()
	var addr string
	if ip := c.RemoteAddr(); ip != nil {
		addr = ip.String()
	}
	if addr != "" {
		log.Infof("handle connection with %s", addr)
		defer log.Infof("close connection with %s", addr)
		r.Counter.OnConnect(addr)
		defer r.Counter.OnDisconnect(addr)
	}

	reader := bufio.NewReaderSize(c, r.readBufferSize)
	for {
		buf, _, err := reader.ReadLine()
		if nil != err {
			if io.EOF != err {
				log.Errorf("tcp communication error %v", err)
			}
			break
		}
		buf_copy := make([]byte, len(buf), len(buf))
		copy(buf_copy, buf)
		k, value, ts, err := m20.ValidatePacket(buf_copy, m20.MediumLegacy, r.validateLevel)
		if err != nil {
			log.Noticef("bad metric format %s, err %v", string(buf_copy), err)
			continue
		}
		key := string(k)
		dp := common.DataPoint{key, value, ts}
		r.DataPointCh <- dp
		for old, new := range r.metricMapping {
			if strings.Contains(key, old) {
				newkey := strings.Replace(key, old, new, 1)
				log.Debugf("metric convertion: [%s]=>[%s]", key, newkey)
				r.DataPointCh <- common.DataPoint{newkey, value, ts}
			}
		}
		if metricAddr, err := common.ExtractMetricAddr(key); err == nil {
			r.Counter.MessageInc(metricAddr, key, int64(ts))
		}
	}
}
